package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.OrderWide;
import com.atguigu.gmall.realtime.bean.PaymentWide;
import com.atguigu.gmall.realtime.bean.ProductStats;
import com.atguigu.gmall.realtime.enums.AppraiseEnum;
import com.atguigu.gmall.realtime.function.DimAsyncFunction;
import com.atguigu.gmall.realtime.utils.ClickhouseUtils;
import com.atguigu.gmall.realtime.utils.DateUtils;
import com.atguigu.gmall.realtime.utils.KafkaUtils;
import com.google.common.collect.Sets;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.Collections;
import java.util.Date;
import java.util.concurrent.TimeUnit;

/**
 * 商品主题统计
 *
 * @author lvbingbing
 * @date 2022-07-04 15:02
 */
@Slf4j
public class ProductStatsApp {
    public static void main(String[] args) throws Exception {
        // 1. 基本环境准备
        // 1.1 流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 1.2 并行度设置
        env.setParallelism(4);
        // 2. 检查点相关设置
        // 2.1 开启检查点
        env.enableCheckpointing(6000L, CheckpointingMode.EXACTLY_ONCE);
        // 2.2 设置检查点超时时间
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        checkpointConfig.setCheckpointTimeout(60000L);
        // 2.3 开启外部化检查点，作业取消时保留检查点
        checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 2.4 设置重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 6000L));
        // 2.5 设置状态后端
        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/ck/gmall"));
        // 2.6 指定操作 hdfs 用户
        System.setProperty("HADOOP_USER_NAME", "atguigu");
        // 3. 从 kafka 读取数据，做结构转换
        String groupId = "product_stats_app_group";
        // 3.1 获取商品点击行为日志数据
        SingleOutputStreamOperator<ProductStats> clickLogDs = getClickLogDs(env, groupId);
        // 3.2 获取商品曝光日志数据
        SingleOutputStreamOperator<ProductStats> displayLogDs = getDisplayLogDs(env, groupId);
        // 3.3 获取商品收藏主题数据
        SingleOutputStreamOperator<ProductStats> favorInfoDs = getFavorInfoDs(env, groupId);
        // 3.4 获取商品加购主题数据
        SingleOutputStreamOperator<ProductStats> cartInfoDs = getCartInfoDs(env, groupId);
        // 3.5 获取订单宽表主题数据
        SingleOutputStreamOperator<ProductStats> orderWideDs = getOrderWideDs(env, groupId);
        // 3.6 获取支付宽表主题数据
        SingleOutputStreamOperator<ProductStats> paymentWideDs = getPaymentWideDs(env, groupId);
        // 3.7 获取退款主题数据
        SingleOutputStreamOperator<ProductStats> refundInfoDs = getRefundInfoDs(env, groupId);
        // 3.8 获取商品评论主题数据
        SingleOutputStreamOperator<ProductStats> commentInfoDs = getCommentInfoDs(env, groupId);
        // 4. 输出是否都能够正常获取数据
        // testLog1(clickLogDs, displayLogDs, favorInfoDs, cartInfoDs);
        // testLog2(orderWideDs, paymentWideDs, refundInfoDs, commentInfoDs);
        // 5. 将不同流的数据合并到一起，指定Watermark和事件时间字段，并做分组、开窗、聚合计算
        SingleOutputStreamOperator<ProductStats> productStatsDs = clickLogDs.union(displayLogDs, favorInfoDs, cartInfoDs, orderWideDs, paymentWideDs, refundInfoDs, commentInfoDs)
                // 指定Watermark和事件时间字段
                .assignTimestampsAndWatermarks(WatermarkStrategy.<ProductStats>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new SerializableTimestampAssigner<ProductStats>() {
                            @Override
                            public long extractTimestamp(ProductStats productStats, long recordTimestamp) {
                                return productStats.getTs();
                            }
                        }))
                // 根据 skuId 进行分组
                .keyBy(ProductStats::getSku_id)
                // 开一个10s的滚动事件时间窗口
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                // 聚合计算
                .reduce(new ReduceFunction<ProductStats>() {
                    @Override
                    public ProductStats reduce(ProductStats productStats1, ProductStats productStats2) throws Exception {
                        return reduceProductStats(productStats1, productStats2);
                    }
                }, new WindowFunction<ProductStats, ProductStats, Long, TimeWindow>() {
                    @Override
                    public void apply(Long aLong, TimeWindow window, Iterable<ProductStats> productStatsList, Collector<ProductStats> out) throws Exception {
                        for (ProductStats productStats : productStatsList) {
                            productStats.setStt(DateUtils.formatDate(new Date(window.getStart())));
                            productStats.setEdt(DateUtils.formatDate(new Date(window.getEnd())));
                            productStats.setTs(System.currentTimeMillis());
                            out.collect(productStats);
                        }

                    }
                });
        // 6. 异步维度关联
        SingleOutputStreamOperator<ProductStats> productStatsDsWithDimInfo = asyncJoinDimInfo(productStatsDs);
        // 输出是否都能够正常获取数据
        // print(productStatsDsWithDimInfo);
        // 7. 将数据写到 clickhouse 中
        String executeSql = "insert into product_stats_2021 values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)";
        productStatsDsWithDimInfo.addSink(ClickhouseUtils.operateClickhouseJdbcSink(executeSql));
        // 8. 触发程序执行
        env.execute();
    }

    @SuppressWarnings("unused")
    private static void print(SingleOutputStreamOperator<ProductStats> productStatsDsWithDimInfo) {
        productStatsDsWithDimInfo.map(s -> {
            log.info("商品主题数据：{}", s);
            return s;
        });
    }

    /**
     * 异步关联维度信息
     *
     * @param productStatsDs 待进行维度关联的商品信息
     * @return 维度关联后的商品信息
     */
    private static SingleOutputStreamOperator<ProductStats> asyncJoinDimInfo(SingleOutputStreamOperator<ProductStats> productStatsDs) {
        // 1. 关联 SKU 维度信息
        SingleOutputStreamOperator<ProductStats> productStatsDsWithSkuInfo = AsyncDataStream.unorderedWait(productStatsDs,
                new DimAsyncFunction<ProductStats>("DIM_SKU_INFO") {
                    @Override
                    public String getDimKey(ProductStats input) {
                        return input.getSku_id().toString();
                    }

                    @Override
                    public void setDimField(ProductStats input, JSONObject dimJsonObj) {
                        input.setSku_name(dimJsonObj.getString("SKU_NAME"));
                        input.setSku_price(dimJsonObj.getBigDecimal("PRICE"));
                        input.setSpu_id(dimJsonObj.getLong("SPU_ID"));
                        input.setCategory3_id(dimJsonObj.getLong("CATEGORY3_ID"));
                        input.setTm_id(dimJsonObj.getLong("TM_ID"));
                    }
                }, 60, TimeUnit.SECONDS);
        // 2. 关联 SPU 维度信息
        SingleOutputStreamOperator<ProductStats> productStatsDsWithSpuInfo = AsyncDataStream.unorderedWait(productStatsDsWithSkuInfo,
                new DimAsyncFunction<ProductStats>("DIM_SPU_INFO") {
                    @Override
                    public String getDimKey(ProductStats input) {
                        return String.valueOf(input.getSpu_id());
                    }

                    @Override
                    public void setDimField(ProductStats input, JSONObject dimJsonObj) {
                        input.setSpu_name(dimJsonObj.getString("SPU_NAME"));
                    }
                }, 60, TimeUnit.SECONDS);
        // 3. 关联品类维度信息
        SingleOutputStreamOperator<ProductStats> productStatsDsWithCategory3Info = AsyncDataStream.unorderedWait(productStatsDsWithSpuInfo,
                new DimAsyncFunction<ProductStats>("DIM_BASE_CATEGORY3") {

                    @Override
                    public String getDimKey(ProductStats input) {
                        return input.getCategory3_id().toString();
                    }

                    @Override
                    public void setDimField(ProductStats input, JSONObject dimJsonObj) {
                        input.setCategory3_name(dimJsonObj.getString("NAME"));
                    }
                }, 60, TimeUnit.SECONDS);
        // 4. 关联品牌维度信息并返回
        return AsyncDataStream.unorderedWait(productStatsDsWithCategory3Info,
                new DimAsyncFunction<ProductStats>("DIM_BASE_TRADEMARK") {
                    @Override
                    public String getDimKey(ProductStats input) {
                        return input.getTm_id().toString();
                    }

                    @Override
                    public void setDimField(ProductStats input, JSONObject dimJsonObj) {
                        input.setTm_name(dimJsonObj.getString("TM_NAME"));
                    }
                }, 60, TimeUnit.SECONDS);
    }

    /**
     * 获取商品点击行为日志数据
     *
     * @param env     <br>
     * @param groupId 消费者组id
     * @return SingleOutputStreamOperator
     */
    private static SingleOutputStreamOperator<ProductStats> getClickLogDs(StreamExecutionEnvironment env, String groupId) {
        String topic = "dwd_page_log";
        return env.addSource(KafkaUtils.getKafkaSource(topic, groupId))
                .filter(jsonStr -> {
                    JSONObject jsonObj = JSON.parseObject(jsonStr);
                    JSONObject pageJsonObj = jsonObj.getJSONObject("page");
                    String pageId = pageJsonObj.getString("page_id");
                    if (StringUtils.isEmpty(pageId)) {
                        log.error("pageId 为空！可能影响页面日志统计分析结果！相关数据：{}", jsonStr);
                        return false;
                    }
                    // 如果当前日志记录的页面是商品的详情页，那么认为这条日志记录的是点击行为
                    return pageId.equals("good_detail");
                })
                .map(jsonStr -> {
                    JSONObject jsonObj = JSON.parseObject(jsonStr);
                    // 获取时间
                    Long ts = jsonObj.getLong("ts");
                    // 获取商品 sku_id
                    JSONObject pageJsonObj = jsonObj.getJSONObject("page");
                    Long skuId = pageJsonObj.getLong("item");
                    // 构建商品主题实体类
                    return ProductStats.builder()
                            .sku_id(skuId)
                            .click_ct(1L)
                            .ts(ts)
                            .build();
                });
    }

    /**
     * 获取商品曝光日志数据
     *
     * @param env     <br>
     * @param groupId 消费者组id
     * @return SingleOutputStreamOperator
     */
    private static SingleOutputStreamOperator<ProductStats> getDisplayLogDs(StreamExecutionEnvironment env, String groupId) {
        String topic = "dwd_display_log";
        return env.addSource(KafkaUtils.getKafkaSource(topic, groupId))
                .filter(jsonStr -> {
                    JSONObject jsonObj = JSON.parseObject(jsonStr);
                    String itemType = jsonObj.getString("item_type");
                    if (StringUtils.isEmpty(itemType)) {
                        log.error("itemType 为空！可能影响曝光日志统计分析结果！相关数据：{}", jsonStr);
                        return false;
                    }
                    // 如果当前 itemType 是 item_id， 说明当前曝光的是商品
                    return "sku_id".equals(itemType);
                })
                .map(jsonStr -> {
                    JSONObject jsonObj = JSON.parseObject(jsonStr);
                    Long skuId = jsonObj.getLong("item");
                    Long ts = jsonObj.getLong("ts");
                    return ProductStats.builder()
                            .sku_id(skuId)
                            .display_ct(1L)
                            .ts(ts)
                            .build();
                });
    }

    /**
     * 获取收藏主题数据
     *
     * @param env     <br>
     * @param groupId 消费者组id
     * @return SingleOutputStreamOperator
     */
    private static SingleOutputStreamOperator<ProductStats> getFavorInfoDs(StreamExecutionEnvironment env, String groupId) {
        String topic = "dwd_favor_info";
        return env.addSource(KafkaUtils.getKafkaSource(topic, groupId))
                .map(jsonStr -> {
                    JSONObject jsonObj = JSON.parseObject(jsonStr);
                    Long skuId = jsonObj.getLong("sku_id");
                    Long timestamp = DateUtils.getTimestamp(jsonObj.getString("create_time"));
                    return ProductStats.builder()
                            .sku_id(skuId)
                            .favor_ct(1L)
                            .ts(timestamp)
                            .build();
                });
    }

    /**
     * 获取商品加购物车主题数据
     *
     * @param env     <br>
     * @param groupId 消费者组id
     * @return SingleOutputStreamOperator
     */
    private static SingleOutputStreamOperator<ProductStats> getCartInfoDs(StreamExecutionEnvironment env, String groupId) {
        String topic = "dwd_cart_info";
        return env.addSource(KafkaUtils.getKafkaSource(topic, groupId))
                .map(jsonStr -> {
                    JSONObject jsonObj = JSON.parseObject(jsonStr);
                    Long skuId = jsonObj.getLong("sku_id");
                    Long timestamp = DateUtils.getTimestamp(jsonObj.getString("create_time"));
                    return ProductStats.builder()
                            .sku_id(skuId)
                            .cart_ct(1L)
                            .ts(timestamp)
                            .build();
                });
    }

    /**
     * 获取订单宽表主题数据
     *
     * @param env     <br>
     * @param groupId 消费者组id
     * @return SingleOutputStreamOperator
     */
    private static SingleOutputStreamOperator<ProductStats> getOrderWideDs(StreamExecutionEnvironment env, String groupId) {
        String topic = "dwm_order_wide";
        return env.addSource(KafkaUtils.getKafkaSource(topic, groupId))
                .map(jsonStr -> {
                    OrderWide orderWide = JSON.parseObject(jsonStr, OrderWide.class);
                    return ProductStats.builder()
                            .sku_id(orderWide.getSku_id())
                            .order_sku_num(orderWide.getSku_num())
                            .order_amount(orderWide.getSplit_total_amount())
                            .orderIdSet(Sets.newHashSet(Collections.singletonList(orderWide.getOrder_id())))
                            .ts(DateUtils.getTimestamp(orderWide.getCreate_time()))
                            .build();
                });
    }

    /**
     * 获取支付宽表主题数据
     *
     * @param env     <br>
     * @param groupId 消费者组id
     * @return SingleOutputStreamOperator
     */
    private static SingleOutputStreamOperator<ProductStats> getPaymentWideDs(StreamExecutionEnvironment env, String groupId) {
        String topic = "dwm_payment_wide";
        return env.addSource(KafkaUtils.getKafkaSource(topic, groupId))
                .map(jsonStr -> {
                    PaymentWide paymentWide = JSON.parseObject(jsonStr, PaymentWide.class);
                    return ProductStats.builder()
                            .sku_id(paymentWide.getSku_id())
                            .payment_amount(paymentWide.getSplit_total_amount())
                            .paidOrderIdSet(Sets.newHashSet(Collections.singletonList(paymentWide.getOrder_id())))
                            .ts(DateUtils.getTimestamp(paymentWide.getCallback_time()))
                            .build();
                });
    }

    /**
     * 获取退款主题数据
     *
     * @param env     <br>
     * @param groupId 消费者组id
     * @return SingleOutputStreamOperator
     */
    private static SingleOutputStreamOperator<ProductStats> getRefundInfoDs(StreamExecutionEnvironment env, String groupId) {
        String topic = "dwd_order_refund_info";
        return env.addSource(KafkaUtils.getKafkaSource(topic, groupId))
                .map(jsonStr -> {
                    JSONObject jsonObj = JSON.parseObject(jsonStr);
                    Long skuId = jsonObj.getLong("sku_id");
                    BigDecimal refundAmount = jsonObj.getBigDecimal("refund_amount");
                    Long orderId = jsonObj.getLong("order_id");
                    Long timestamp = DateUtils.getTimestamp(jsonObj.getString("create_time"));
                    return ProductStats.builder()
                            .sku_id(skuId)
                            .refund_amount(refundAmount)
                            .refundOrderIdSet(Sets.newHashSet(Collections.singletonList(orderId)))
                            .ts(timestamp)
                            .build();
                });
    }

    /**
     * 获取评论主题数据
     *
     * @param env     <br>
     * @param groupId 消费者组id
     * @return SingleOutputStreamOperator
     */
    private static SingleOutputStreamOperator<ProductStats> getCommentInfoDs(StreamExecutionEnvironment env, String groupId) {
        String topic = "dwd_comment_info";
        return env.addSource(KafkaUtils.getKafkaSource(topic, groupId))
                .map(jsonStr -> {
                    JSONObject jsonObj = JSON.parseObject(jsonStr);
                    Long skuId = jsonObj.getLong("sku_id");
                    String appraise = jsonObj.getString("appraise");
                    Long timestamp = DateUtils.getTimestamp(jsonObj.getString("create_time"));
                    return ProductStats.builder()
                            .sku_id(skuId)
                            .comment_ct(1L)
                            .good_comment_ct(AppraiseEnum.getByCode(appraise) == AppraiseEnum.GOOD ? 1L : 0L)
                            .ts(timestamp)
                            .build();
                });
    }

    private static ProductStats reduceProductStats(ProductStats productStats1, ProductStats productStats2) {
        productStats1.setDisplay_ct(productStats1.getDisplay_ct() + productStats2.getDisplay_ct());
        productStats1.setClick_ct(productStats1.getClick_ct() + productStats2.getClick_ct());
        productStats1.setCart_ct(productStats1.getCart_ct() + productStats2.getCart_ct());
        productStats1.setFavor_ct(productStats1.getFavor_ct() + productStats2.getFavor_ct());
        productStats1.setOrder_amount(productStats1.getOrder_amount().add(productStats2.getOrder_amount()));
        productStats1.getOrderIdSet().addAll(productStats2.getOrderIdSet());
        productStats1.setOrder_ct((long) productStats1.getOrderIdSet().size());
        productStats1.setOrder_sku_num(productStats1.getOrder_sku_num() + productStats2.getOrder_sku_num());
        productStats1.setPayment_amount(productStats1.getPayment_amount().add(productStats2.getPayment_amount()));

        productStats1.getRefundOrderIdSet().addAll(productStats2.getRefundOrderIdSet());
        productStats1.setRefund_order_ct((long) productStats1.getRefundOrderIdSet().size());
        productStats1.setRefund_amount(productStats1.getRefund_amount().add(productStats2.getRefund_amount()));

        productStats1.getPaidOrderIdSet().addAll(productStats2.getPaidOrderIdSet());
        productStats1.setPaid_order_ct((long) productStats1.getPaidOrderIdSet().size());

        productStats1.setComment_ct(productStats1.getComment_ct() + productStats2.getComment_ct());
        productStats1.setGood_comment_ct(productStats1.getGood_comment_ct() + productStats2.getGood_comment_ct());

        return productStats1;
    }

    /**
     * 输出是否都能够正常获取数据
     *
     * @param clickLogDs   商品点击行为日志数据
     * @param displayLogDs 商品曝光日志数据
     * @param favorInfoDs  商品收藏主题数据
     * @param cartInfoDs   商品加购主题数据
     */
    @SuppressWarnings("unused")
    private static void testLog1(SingleOutputStreamOperator<ProductStats> clickLogDs, SingleOutputStreamOperator<ProductStats> displayLogDs,
                                 SingleOutputStreamOperator<ProductStats> favorInfoDs, SingleOutputStreamOperator<ProductStats> cartInfoDs) {
        clickLogDs.map(s -> {
            log.info("商品点击日志：{}", s);
            return s;
        });
        displayLogDs.map(s -> {
            log.info("商品曝光日志：{}", s);
            return s;
        });
        favorInfoDs.map(s -> {
            log.info("商品收藏数据：{}", s);
            return s;
        });
        cartInfoDs.map(s -> {
            log.info("商品加购数据：{}", s);
            return s;
        });
    }

    /**
     * 输出是否都能够正常获取数据
     *
     * @param orderWideDs   订单宽表主题数据
     * @param paymentWideDs 支付宽表主题数据
     * @param refundInfoDs  退款主题数据
     * @param commentInfoDs 商品评论主题数据
     */
    @SuppressWarnings("unused")
    private static void testLog2(SingleOutputStreamOperator<ProductStats> orderWideDs, SingleOutputStreamOperator<ProductStats> paymentWideDs,
                                 SingleOutputStreamOperator<ProductStats> refundInfoDs, SingleOutputStreamOperator<ProductStats> commentInfoDs) {

        orderWideDs.map(s -> {
            log.info("订单宽表数据：{}", s);
            return s;
        });
        paymentWideDs.map(s -> {
            log.info("支付宽表数据：{}", s);
            return s;
        });
        refundInfoDs.map(s -> {
            log.info("退款数据：{}", s);
            return s;
        });
        commentInfoDs.map(s -> {
            log.info("评论数据：{}", s);
            return s;
        });
    }
}
